﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.IO;
using JetBlack.IO;
using System.Net.Sockets;

namespace JetBlack.Net.Sockets
{
    internal class QueuingFrameStream : FrameStream
    {
        private readonly Queue<byte[]> readQueue = new Queue<byte[]>();
        private byte[] readDataBuffer = null;
        private readonly byte[] readLengthBuffer = new byte[4];
        private int readOffset = 0;

        public QueuingFrameStream(Stream stream)
            : base(stream)
        {
        }

        public bool IsReadQueueEmpty
        {
            get { lock (readQueue) { return readQueue.Count == 0; } }
        }

        public byte[] Pop()
        {
            lock (readQueue) { return readQueue.Dequeue(); }
        }

        public bool ReadUntilBlocked()
        {
            lock (readQueue)
            {
                int count = readQueue.Count;

                if (readDataBuffer == null)
                {
                    readOffset += ReadNonBlocking(readLengthBuffer, readOffset);

                    if (readOffset == readLengthBuffer.Length)
                    {
                        readOffset = 0;
                        readDataBuffer = new byte[NetworkBitConverter.ToInt32(readLengthBuffer, 0)];
                    }
                }

                if (readDataBuffer != null)
                {
                    readOffset += ReadNonBlocking(readDataBuffer, readOffset);

                    if (readOffset == readDataBuffer.Length)
                    {
                        readQueue.Enqueue(readDataBuffer);
                        readOffset = 0;
                        readDataBuffer = null;
                    }
                }

                return readQueue.Count > count;
            }
        }

        private int ReadNonBlocking(byte[] buf, int offset)
        {
            try
            {
                int bytesRead = InnerStream.Read(buf, offset, buf.Length - offset);
                return bytesRead;
            }
            catch (IOException ioException)
            {
                if (ioException.InnerException is SocketException)
                {
                    var socketException = (SocketException)ioException.InnerException;
                    var errorCode = (SocketError)socketException.ErrorCode;
                    if (errorCode == SocketError.WouldBlock || errorCode == SocketError.TimedOut)
                        return 0;
                }
                throw;
            }
        }

        private readonly Queue<byte[]> writeQueue = new Queue<byte[]>();
        private byte[] writeDataBuffer = null;
        private byte[] writeLengthBuffer = null;

        public void Push(byte[] bytes)
        {
            lock (writeQueue)
            {
                if (writeDataBuffer == null)
                {
                    writeDataBuffer = bytes;
                    writeLengthBuffer = NetworkBitConverter.GetBytes(writeDataBuffer.Length);
                }
                else
                    writeQueue.Enqueue(bytes);
            }
        }

        public bool IsWriteQueueEmpty
        {
            get { lock (writeQueue) { return writeDataBuffer == null; } }
        }

        public bool WriteUntilBlocked()
        {
            lock (writeQueue)
            {
                while (writeLengthBuffer != null || writeDataBuffer != null || writeQueue.Count > 0)
                {
                    try
                    {
                        if (writeLengthBuffer != null)
                        {
                            Write(writeLengthBuffer, 0, writeLengthBuffer.Length);
                            writeLengthBuffer = null;
                        }

                        if (writeDataBuffer != null)
                        {
                            Write(writeDataBuffer, 0, writeDataBuffer.Length);
                            if (writeQueue.Count == 0)
                                writeDataBuffer = null;
                            else
                            {
                                writeDataBuffer = writeQueue.Dequeue();
                                writeLengthBuffer = NetworkBitConverter.GetBytes(writeDataBuffer.Length);
                            }
                        }
                    }
                    catch (IOException ioException)
                    {
                        if (ioException.InnerException is SocketException)
                        {
                            var socketException = (SocketException)ioException.InnerException;
                            var errorCode = (SocketError)socketException.ErrorCode;
                            if (errorCode == SocketError.WouldBlock || errorCode == SocketError.TimedOut)
                                break;
                        }
                        throw;
                    }
                }

                return writeDataBuffer == null;
            }
        }
    }
}
